import pandas as pd
import fbprophet
from concurrent.futures import ProcessPoolExecutor, as_completed
import time

# 经运行得出的最佳参数组合为seasonality_prior_scale=0.001，changepoint_prior_scale=0.01


def wmape_operator(predict_data, check_data):
    """
    计算wmape1、wmape2，即wmape的分子和分母
    :param predict_data: 预测数据集，包含日期('ds')、预测销量('yhat')
    :param check_data: 验证数据集
    :return: wmape1、wmape2值
    """
    wmape1 = 0
    wmape2 = 0
    for i in check_data.iterrows():
        bias = abs(float(predict_data.loc[predict_data['ds'] == i[1].ds]['yhat']) - i[1].y)
        wmape1 += bias
        wmape2 += i[1].y
    return wmape1, wmape2


def predictor(data, period, holiday, seasonality_prior_scale, changepoint_prior_scale, holidays_prior_scale=10):
    """
    Prophet算法预测模型
    :param data: 商品销量数据，包含日期('ds')、销量('y')
    :param period: 预测时期长度
    :return: 商品销量预测数据，包含日期('ds')、预测销量('yhat')
    """
    # 模型构建
    m = fbprophet.Prophet(interval_width=0.95,  # 类似于置信区间的置信度
                          weekly_seasonality=True,  # 加入周季节性趋势
                          seasonality_prior_scale=seasonality_prior_scale,  # 季节性组件的强度
                          holidays_prior_scale=holidays_prior_scale,  # 节假日模型组件的强度
                          changepoint_prior_scale=changepoint_prior_scale,  # 调节“change point”选择的灵活度
                          seasonality_mode='additive',  # 加法季节性
                          holidays=holiday
                          )

    # 依据日期数加入趋势项
    days = len(data)
    if days > 240:
        m.add_seasonality(name='quarterly', period=91.25, fourier_order=8, mode='additive')  # 每季度 'multiplicative'
    if days > 60:
        m.add_seasonality(name='monthly', period=30.5, fourier_order=5, mode='additive')  # 每月 'multiplicative'

    # 数据拟合与模型预测
    m.fit(data)
    future = m.make_future_dataframe(periods=period)
    forecast = m.predict(future)
    forecast['yhat'] = forecast['yhat'].apply(lambda x: (0 if x < 0 else x))
    return forecast[['ds', 'yhat']]


if __name__ == '__main__':
    # 导入数据，并修改日期、商品ID、是否促销列的数据格式
    train_data = pd.read_csv('train.csv', dtype={'sale_date': str, 'goodsid': int, 'is_pro': int})
    test_data = pd.read_csv('test.csv', dtype={'sale_date': str, 'goodsid': int, 'is_pro': int})
    goods_info = pd.read_csv('goods_ch.csv',
                             usecols=['goodsid', 'div_id', 'catg_l_id',
                                      'catg_m_id', 'catg_s_id', 'season_class'])
    train_data['sale_date'] = pd.to_datetime(train_data['sale_date'], format='%Y%m%d')
    test_data['sale_date'] = pd.to_datetime(test_data['sale_date'], format='%Y%m%d')
    date_ch = pd.read_csv('date_ch.csv', encoding='gbk')
    # 节假日处理
    holiday1 = date_ch[['dim_date_id', 'official_holiday_name']].dropna().reset_index(drop=True)
    holiday1.columns = ['ds', 'holiday']
    holiday2 = date_ch[['dim_date_id', 'festival_name']].dropna().reset_index(drop=True)
    holiday2.columns = ['ds', 'holiday']
    holiday_plus_ls = ['情人节', '元宵节', '七夕节', '教师节', '圣诞节', '平安夜']
    holiday2 = holiday2[holiday2['holiday'].isin(holiday_plus_ls)].reset_index(drop=True)
    holiday = pd.concat([holiday1, holiday2], ignore_index=True).sort_values('ds')
    holiday['ds'] = pd.to_datetime(holiday['ds'], format='%Y%m%d')

    param_grid = {
        'seasonality_prior_scale': [0.001, 0.01, 0.1, 0.5],
        'changepoint_prior_scale': [0.01, 0.1, 1.0, 10.0],
    }
    result_list = {}
    for sps in param_grid['seasonality_prior_scale']:
        for cps in param_grid['changepoint_prior_scale']:
            result_list[(sps, cps)] = []
            t1 = time.time()
            with ProcessPoolExecutor(max_workers=4) as executor:
                wmape1 = 0
                wmape2 = 0
                futures = {}
                for gid in goods_info['goodsid']:
                    goods_data = train_data.loc[train_data['goodsid'] == gid][['sale_date', 'sales_qty']]
                    goods_data = goods_data.reset_index(drop=True)
                    goods_data.columns = ['ds', 'y']
                    # 找到有销量日期的训练集与验证集分割点，保证验证集销量数据不全为0
                    day_num = len(goods_data)
                    split_date = goods_data.iloc[int(day_num * 0.85), 0]
                    # 消除训练集部分中异常值(异常高的销量修改为None)
                    mean = goods_data['y'].mean()
                    std_15 = goods_data['y'].std() * 15
                    goods_data.loc[(goods_data['y'] > mean + std_15) & (goods_data['ds'] <= split_date), 'y'] = None
                    # 补全销量为0的日期销量数据
                    start = goods_data['ds'].min()
                    end = test_data[test_data['goodsid'] == gid]['sale_date'].min()
                    goods_data = goods_data.set_index('ds')
                    all_date = pd.date_range(start=start, end=end)[:-1]
                    goods_data = goods_data.reindex(all_date, fill_value=0)
                    goods_data['ds'] = goods_data.index
                    goods_data = goods_data.reset_index(drop=True)
                    # 划分训练集与验证集
                    goods_train_data = goods_data[goods_data['ds'] <= split_date]
                    goods_check_data = goods_data[goods_data['ds'] > split_date].reset_index(drop=True)
                    period = len(goods_check_data)
                    # 放入进程池，并记录相应的商品编号
                    job = executor.submit(predictor,
                                          goods_train_data,
                                          period,
                                          holiday,
                                          sps, cps)
                    futures[job] = (gid, goods_check_data)
                for job in as_completed(futures):
                    predict_data = job.result()
                    gid, goods_check_data = futures[job]
                    a, b = wmape_operator(predict_data, goods_check_data)
                    wmape1 += a
                    wmape2 += b
                result_list[(sps, cps)].append("wmape: {}".format(wmape1 / wmape2))
            result_list[(sps, cps)].append('共耗时: {}'.format(time.time() - t1))
            print(result_list[(sps, cps)][0], result_list[(sps, cps)][1])
            print('-'*100)
